package server

import (
	"log"
	"net"
	"proxyRelayServer/src/util"
	"strings"
	"sync"
	"time"
)

type RelayBroker struct {
	ListenAddr net.Addr
	SubAddr    net.Addr
	WhiteList  map[string]struct{}
	Lock       *sync.RWMutex
}

func NewRelayBroker(
	listenAddr net.Addr, subAddr net.Addr,
) *RelayBroker {
	return &RelayBroker{listenAddr, subAddr, make(map[string]struct{}), &sync.RWMutex{}}
}

func (broker *RelayBroker) isWhiteAddr(addr net.Addr) bool {
	broker.Lock.RLock()
	defer broker.Lock.RUnlock()
	index := strings.LastIndex(addr.String(), ":")
	if index == -1 {
		return false
	}
	if len(broker.WhiteList) == 0 {
		return false
	}
	_, ok := broker.WhiteList[addr.String()[:index]]
	return ok
}

func (broker *RelayBroker) Listen() (err error) {
	listener, err := net.Listen("tcp", broker.ListenAddr.String())
	if err != nil {
		return err
	}
	pub, err := net.Listen("tcp", broker.SubAddr.String())
	if err != nil {
		listener.Close()
		return err
	}
	//订阅连接缓存队列
	pubBacklog := util.NewBlockingQueue[net.Conn]()
	go func() {
		for {
			//监听订阅地址 接收tcp连接
			subConn, err := pub.Accept()
			if err != nil {
				log.Println(err)
				continue
			}

			go func() {
				log.Println("sub conn", subConn.RemoteAddr())
				checkLimit := 5
				checkCount := 0
			recheck:
				if !broker.isWhiteAddr(subConn.RemoteAddr()) {
					if checkCount < checkLimit {
						checkCount++
						time.Sleep(5 * time.Second)
						log.Println("重新检查", subConn.RemoteAddr(), "是否在白名单")
						goto recheck
					} else {
						log.Println(subConn.RemoteAddr(), "不在白名单")
						subConn.Close()
						return
					}
				} else {
					log.Println("subConn:", subConn.RemoteAddr(), "在白名单")
				}
				conn := subConn.(*net.TCPConn)
				err = conn.SetKeepAlive(true)
				if err != nil {
					log.Println(err)
					return
				}
				err = conn.SetKeepAlivePeriod(time.Minute)
				if err != nil {
					log.Println(err)
					return
				}
				//将订阅连接放入缓存队列
				pubBacklog.Append(subConn)
			}()
		}
	}()
	userConnBacklogDeque := util.NewBlockingDeque[net.Conn]()
	go func() {
		for range time.Tick(time.Millisecond * 5) { //每五毫秒扫描一次 用户连接的backlog (videojs拖动时会有很大的并发量,对其延迟处理能更好的提供服务)
			for {
				userConn, ok := userConnBacklogDeque.TakeFront()
				if !ok {
					//log.Println("empty userConnBacklogDeque")
					break
				}
				log.Println(userConn.RemoteAddr())
				if !broker.isWhiteAddr(userConn.RemoteAddr()) {
					log.Println(userConn.RemoteAddr(), "不在白名单")
					userConn.Close()
					continue
				}
				go func() {
					pop := pubBacklog.Pop()
					defer func() {
						if err := recover(); err != nil {
							log.Println(err)
						}
						_ = userConn.Close()
						//关闭占用的连接
						_ = pop.Close()
					}()
					group := sync.WaitGroup{}
					group.Add(1)
					go func() {
						defer group.Done()
						written, err := util.CopyBuf(userConn, pop, time.Second*30, time.Second*300)
						if err != nil {
							log.Println(err)
							err := userConn.Close()
							if err != nil {
								log.Println(err)
							}
						}
						log.Printf("%s -> %s %d bytes", pop.RemoteAddr(), userConn.RemoteAddr(), written)
					}()
					written, err := util.CopyBuf(pop, userConn, time.Second*300, time.Second*30)
					if err != nil {
						log.Println(err)
					}
					log.Printf("%s -> %s %d bytes", userConn.RemoteAddr(), pop.RemoteAddr(), written)
					//等待异步读写结束
					group.Wait()
				}()
			}
		}
	}()
	for {
		userConn, err := listener.Accept()
		if err != nil {
			log.Println(err)
			continue
		}
		userConnBacklogDeque.PushFront(userConn)
	}
}

func peek(conn net.Conn) ([]byte, error) {
	peekBuf := make([]byte, 4)
	n, err := conn.Read(peekBuf)
	if err != nil {
		return nil, err
	}
	return peekBuf[:n], nil
}
